package com.shujia.flink.core

import org.apache.flink.streaming.api.datastream.DataStreamSink
import org.apache.flink.streaming.api.scala._

object Demo4Parallelism {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      *
      * flink 的并行度
      * 1、如果代码中不设置并行度，在提交任务的是偶默认是1，  可以在提交任务的是偶指定并行度 参数：-p
      * 2、代码中可以设置并行度 setParallelism   ，代码的优先级比-p参数幺高
      * 3、每一个节点可以单独设置并行度
      * 4、如果节点之前的关系没有产生shuffle 并且并行度一致，可以合并成一个task
      *
      *
      */

    /**
      * 设置默认的并行度
      *
      */
    //env.setParallelism(4)

    /**
      * 读取socket的并行度只能是1
      *
      */

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    linesDS.name("读取socket数据")

    println(s"linesDS的并行度：${linesDS.parallelism}")

    /**
      * 节点的并行度由默认并行度决定
      * 也可以单独设置每一个节点的并行度
      *
      */

    val wordsDS: DataStream[String] = linesDS.flatMap(_.split(","))

    //单独设置每一个节点的并行度
    //wordsDS.setParallelism(1)

    wordsDS.name("展开数据")

    println(s"wordsDS的并行度：${wordsDS.parallelism}")

    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))

    /**
      * 如果上游节点和下游节点并行度一样，同时算子不产生shuffle, 可以合并成一个task
      *
      */

    // kvDS.setParallelism(1)

    kvDS.name("转换成kv")

    println(s"kvDS的并行度：${kvDS.parallelism}")
    /**
      * keyBy不能设置名称和并行度
      *
      */

    val keyVyDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    val countDS: DataStream[(String, Int)] = keyVyDS.sum(1)

    countDS.name("统计单词的数量")
    //countDS.setParallelism(2)


    val print: DataStreamSink[(String, Int)] = countDS.print()

    //print.setParallelism(2)
    print.name("打印结果")

    env.execute()

  }
}
